package com.lvmama.rhino.analyze.processing

import com.lvmama.rhino.common.utils.JDBCUtil.Conversions
import com.lvmama.rhino.common.utils.Utils
import org.apache.spark.SparkContext
import org.apache.spark.sql.DataFrame
import org.apache.spark.sql.functions._

/**
 * Created by shibing on 2018/05/02.
 */
object SuperFreeTravel_Hp{
  def process(sc: SparkContext, pageForward: DataFrame): Unit = {
    import com.lvmama.rhino.analyze.client.SuperFreeTravel._

    val group = sc.broadcast(Seq("pageTypeCode", "channelCode", "platformCode"))

    val commons = Seq("deviceToken", "sessionId", "channelCode", "platformCode", "pageTypeCode", "timestamp", "pageParam", "buttonCode")

    //过滤跳转事件
    val homepageData=pageForward.filter(col("logType")==="click").filter(col("pageTypeCode") isin ("3tqxx",
       "3tqxxsh3",
      "3tqxxtng",
       "3tqxx4ot",
       "3tqxxtnggwh",
       "3tqxxtnge1g",
       "3tqxxtngfdo",
       "3tqxxl4s",
      "3tqxx909",
      "3tqxx02m",
      "3tqxxiax"
    ))
//    homepageData.show(250,false)
//    println("----------pageFarward end------------------")
//    val homepageDataCount= homepageData.count()
//    println("----------countnum=--------------" + homepageDataCount.toString)

    //从PageParam中拆分recommendId 和recommendName
   val getRecommendInfo = homepageData.select(commons.map(col):_*)
      .withColumn("recommendId"
        ,when(col("pageParam").getItem("recommendId").isNotNull, col("pageParam").getItem("recommendId")).otherwise(""))
      .withColumn("recommendName"
        ,when(col("pageParam").getItem("recommendName").isNotNull, col("pageParam").getItem("recommendName")).otherwise(""))
      .withColumn("tagId"
        ,when(col("pageParam").getItem("tagId").isNotNull, col("pageParam").getItem("tagId")).otherwise(""))

//    getRecommendInfo.show(100,false)
//    println("--------------------------getRecommendInfo-----------------")

    //
    val getPagenameButtonname = getRecommendInfo.withColumn("pageName", when(col("pageTypeCode")==="3tqxx", "首页")
        .when(col("pageTypeCode")==="3tqxxsh3", "玩法列表")
        .when(col("pageTypeCode")==="3tqxxtng", "选择资源页")
        .when(col("pageTypeCode")==="3tqxx4ot", "玩法详情页")
        .when(col("pageTypeCode")==="3tqxxtnggwh", "更换交通页")//更换返程页
        .when(col("pageTypeCode")==="3tqxxtnge1g", "酒店列表页")
        .when(col("pageTypeCode")==="3tqxxtngfdo", "酒店详情页")
        .when(col("pageTypeCode")==="3tqxxl4s", "选择城市页")
        .when(col("pageTypeCode")==="3tqxx909", "选择日期页面")
        .when(col("pageTypeCode")==="3tqxx02m", "选择出游日期和人数页面")
        .when(col("pageTypeCode")==="3tqxxiax", "填写订单页")
      ).withColumn("buttonName",
      //首页
        when(col("buttonCode")==="search", "搜索")
        .when(col("buttonCode")==="searchHistory","查询记录")
        .when(col("buttonCode")==="moreRecommend" && col("pageTypeCode")==="3tqxx","更多玩法")
      //buttCode为空时，这样取值
        .when(col("buttonCode").isNull || col("buttonCode")==="", concat_ws("_",format_string("玩法推荐"),col("recommendId"),col("recommendName")))
      //行程向导页
        .when(col("buttonCode")==="filter" && col("tagId").contains("days") && col("tagId").contains("theme"),"出游天数_筛选")
        .when(col("buttonCode")==="filter" && col("tagId").contains("theme"),"筛选")
        .when(col("buttonCode")==="filter" && col("tagId").contains("days"),"出游天数")
      //选择资源页
        .when(col("buttonCode")==="moreRecommend" && col("pageTypeCode")==="3tqxxtng", "展开玩法")
        .when(col("buttonCode")==="changeTrip" ,"更改行程")
        .when(col("buttonCode")==="changePlane" ,"更改机票")
        .when(col("buttonCode")==="deletePlane" ,"删除机票")
        .when(col("buttonCode")==="changeRoom" ,"更换房型")
        .when(col("buttonCode")==="roomNum" ,"房间数量")
        .when(col("buttonCode")==="changeHotel" ,"更换酒店")
        .when(col("buttonCode")==="deleteHotel" ,"删除酒店")
        .when(col("buttonCode")==="addResource" ,"添加资源")
        .when(col("buttonCode")==="next" ,"下一步")
      //玩法详情页
        .when(col("buttonCode")==="moreCity", "更多出发地")
        .when(col("buttonCode")==="hotelDetail", "查看酒店")
        .when(col("buttonCode")==="reserve", concat_ws("_",format_string("立即预定"),col("recommendId"),col("recommendName")))
      //更换交通页
        .when(col("buttonCode")==="moreSpace","查看更多舱位")
        .when(col("buttonCode")==="timeSort","时间")
        .when(col("buttonCode")==="priceSort","价格")
        .when(col("buttonCode")==="forwardPlane","去程机票")
        .when(col("buttonCode")==="filter" && col("pageTypeCode")==="3tqxxtnggwh","筛选")
      //酒店列表页
        .when(col("buttonCode")==="stayDate","入住时间")
        .when(col("buttonCode")==="chooseHotel","酒店")
        .when(col("buttonCode")==="sort", concat_ws("_",format_string("排序"),split(col("tagId"),"\"").getItem(3)))
        .when(col("buttonCode")==="filter" && col("pageTypeCode")==="3tqxxtnge1g","筛选")
      //酒店详情页
        .when(col("buttonCode")==="comm","点评")
        .when(col("buttonCode")==="details","详情")
        .when(col("buttonCode")==="map","地图")
//        .when(col("buttonCode")==="stayDate","入住时间")
        .when(col("buttonCode")==="chooseRoom","选择房型")
//        .when(col("buttonCode")==="roomNum","房间数")
      //选择城市页
//        .when(col("buttonCode")==="search","搜索")
        .when(col("buttonCode")==="hotCity","热门城市")
      //选择日期页面
        .when(col("buttonCode")==="forwardDate","去程日期")
        .when(col("buttonCode")==="backDate","返程日期")
      //选择出游日期和人数页面
        .when(col("buttonCode")==="choosedate","选择日期")
        .when(col("buttonCode")==="tips","提示")
        .when(col("buttonCode")==="adults","选择成人数")
        .when(col("buttonCode")==="kids","选择儿童数")
       // .when(col("buttonCode")==="next","下一步")
      //填写订单页
        .when(col("buttonCode")==="submit", "提交订单")

      )

//    getPagenameButtonname.show(70,false)
//    println("-----------------getPagenameButtonname--------------")

    //统计玩法列表出游天数筛选结果
    val aggCols=Seq("pageTypeCode","pageName","buttonName")
    val getTravelDays=getPagenameButtonname.filter(col("pageTypeCode")==="3tqxxsh3").filter(col("buttonName").contains("出游天数"))
        .select(col("pageTypeCode"),col("pageName"),when(col("buttonName").isNotNull,"出游天数") as "buttonName")
          .groupBy(aggCols.map(col(_)): _*).agg(count("*") as "amount")
//    getTravelDays.show(50,false)
//    println("-------------getTravelDays------------------")

    //统计玩法列表主题筛选结果
    val getTheme=getPagenameButtonname.filter(col("pageTypeCode")==="3tqxxsh3").filter(col("buttonName").contains("筛选"))
      .select(col("pageTypeCode"),col("pageName"),when(col("buttonName").isNotNull,"筛选") as "buttonName")
      .groupBy(aggCols.map(col(_)): _*).agg(count("*") as "amount")
//    getTheme.show(50,false)
//    println("-------------getTheme------------------")

    //其他页面的统计结果
    val otherPagesStatis =getPagenameButtonname.filter(col("pageTypeCode")=!="3tqxxsh3")
      .select(aggCols.map(col(_)): _*).groupBy(aggCols.map(col(_)): _*).agg(count("*") as "amount")

//    otherPagesStatis.show(20,false)
//    println("---------------otherPagesStatis-------------")

    //合并
    val mergeResult=otherPagesStatis.union(getTheme).union(getTravelDays)
      .withColumn("elem_type",
       concat_ws("_",format_string("超级自由行"),col("pageName"),col("buttonName")))

//    mergeResult.show(20,false)
//    println("---------------mergeResult-------------")

    //获取当前时间的前一天日期
    val yesterday = Utils.getYesterday()

    import Conversions._

    //数据入库
    mergeResult.select(col("pageTypeCode").as("page_code"),col("elem_type"), col("amount"))
      .withColumn("oper_date",lit(yesterday))
      .coalesce(20)
      .insertDF2MysqlDirect("flow_superfreetravel")

    //统计页面pv uv
    val homepageVisitCounts =homepageData
      .transform(groupByColumnsPvUv(group))
      .coalesce(20)
      .withColumnRenamed("counts", "homepage_visit_counts")
      .withColumnRenamed("uvCounts", "homepage_uv_counts")

    //数据入库
     homepageVisitCounts.select(col("channelCode").as("channel_code"), col("platformCode").as("platform_code"),
        col("pageTypeCode").as("page_code"), col("homepage_visit_counts").as("amount"),
        col("homepage_uv_counts").as("amount_uv"))
      .withColumn("oper_date", lit(yesterday))
      .coalesce(20)
      .insertDF2MysqlDirect("flow_superfreetravel_amount")

  }
}
